﻿/*************************************
 * Creator:peiyang
 * DateTime:2021/3/18 8:25:05
 * Description:<Description>
 * CopyRight:臭皮匠工作室
 * ***********************************/

using Apache.NMS;
using Apache.NMS.ActiveMQ;
using Cpjit.SunnyPaine.Toolkit.ActiveMQ.Enums;
using Cpjit.SunnyPaine.Toolkit.ActiveMQ.Exceptions;
using Cpjit.SunnyPaine.Toolkit.ActiveMQ.Model;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Reflection;
using System.Text;
using System.Threading.Tasks;

namespace Cpjit.SunnyPaine.Toolkit.ActiveMQ.Impl
{
    /// <summary>
    /// 提供ActiveMQ客户端工具。
    /// </summary>
    public class ActivemqClient : IActivemqClient
    {
        #region 私有变量
        /// <summary>
        /// 连接对象。
        /// </summary>
        private IConnection connection;
        /// <summary>
        /// 创建连接的工厂。
        /// </summary>
        private IConnectionFactory connectionFactory;
        /// <summary>
        /// 会话对象。
        /// </summary>
        private ISession session;
        /// <summary>
        /// 表示可由多个线程同时访问的DestType.Queue消息模式的生产者键值对线程安全集合。
        /// <para>键：消息目标名称；值：消息生产者。</para>
        /// </summary>
        private ConcurrentDictionary<string, IMessageProducer> producerQueueDict;
        /// <summary>
        /// 表示可由多个线程同时访问的DestType.Topic消息模式的生产者键值对线程安全集合。
        /// <para>键：消息目标名称；值：消息生产者。</para>
        /// </summary>
        private ConcurrentDictionary<string, IMessageProducer> producerTopicDict;
        /// <summary>
        /// 表示可由多个线程同时访问的DestType.Queue消息模式的消费者键值对线程安全集合。
        /// <para>键：消息目标名称；值：消息消费者。</para>
        /// </summary>
        private ConcurrentDictionary<string, IMessageConsumer> consumerQueueDict;
        /// <summary>
        /// 表示可由多个线程同时访问的DestType.Topic消息模式的消费者键值对线程安全集合。
        /// <para>键：消息目标名称；值：消息消费者。</para>
        /// </summary>
        private ConcurrentDictionary<string, IMessageConsumer> consumerTopicDict;
        #endregion


        #region 公共属性，IActivemqClient成员
        /// <summary>
        /// ActiveMQ代理服务的地址（必填项）。
        /// </summary>
        public string BrokerUri { get; set; }

        /// <summary>
        /// ActiveMQ代理服务的地址（仅包含IP地址。）
        /// </summary>
        public string BrokerUriIpAddress { get; private set; }

        /// <summary>
        /// 用户名。
        /// </summary>
        public string UserName { get; set; }

        /// <summary>
        /// 密码。
        /// </summary>
        public string Password { get; set; }

        /// <summary>
        /// 客户端Id名称。
        /// </summary>
        public string ClientId { get; set; }

        /// <summary>
        /// 发送消息持久化的有效时长。默认2秒。
        /// </summary>
        public TimeSpan LiveTime { get; set; } = new TimeSpan(0, 0, 2);

        /// <summary>
        /// 是否连接到ActiveMQ。
        /// </summary>
        public bool IsConnected { get; private set; }

        /// <summary>
        /// 字符编码。默认UTF8。
        /// </summary>
        public Encoding Encoding { get; set; } = Encoding.UTF8;
        #endregion


        #region 委托,IActivemqClient成员
        /// <summary>
        /// 当与服务端连接状态变化时发生。
        /// </summary>
        public event Action<IActivemqClient, StatusEventArgs> ConnectionStatusChanged;

        /// <summary>
        /// 当出现错误时发生。
        /// </summary>
        public event Action<IActivemqClient, ExceptionEventArgs> Error;
        #endregion


        #region 构造方法
        /// <summary>
        /// 使用默认的参数创建实例。
        /// </summary>
        public ActivemqClient() : this(null)
        { }

        /// <summary>
        /// 使用指定的参数创建实例。
        /// </summary>
        /// <param name="brokerUri"></param>
        public ActivemqClient(string brokerUri)
            : this(brokerUri, null, null)
        { }

        /// <summary>
        /// 使用指定的参数创建实例。
        /// </summary>
        /// <param name="brokerUri"></param>
        /// <param name="userName"></param>
        /// <param name="password"></param>
        public ActivemqClient(string brokerUri, string userName, string password)
            : this(brokerUri, userName, password, null)
        { }

        /// <summary>
        /// 使用指定的参数创建实例。
        /// </summary>
        /// <param name="brokerUri"></param>
        /// <param name="userName"></param>
        /// <param name="password"></param>
        /// <param name="clientId"></param>
        public ActivemqClient(string brokerUri, string userName, string password, string clientId)
        {
            this.producerQueueDict = new ConcurrentDictionary<string, IMessageProducer>();
            this.producerTopicDict = new ConcurrentDictionary<string, IMessageProducer>();
            this.consumerQueueDict = new ConcurrentDictionary<string, IMessageConsumer>();
            this.consumerTopicDict = new ConcurrentDictionary<string, IMessageConsumer>();

            this.BrokerUri = brokerUri;
            this.UserName = userName;
            this.Password = password;
            this.ClientId = clientId;

            if (brokerUri.StartsWith("activemq"))
            {
                int startIndex = brokerUri.IndexOf("activemq:") + "activemq:".Length;
                int endIndex = brokerUri.Contains("?") ? brokerUri.IndexOf("?") - "activemq:".Length : brokerUri.Length;
                this.BrokerUriIpAddress = brokerUri.Substring(startIndex, endIndex).Replace("(", string.Empty).Replace(")", string.Empty);
            }
            else if (brokerUri.StartsWith("failover"))
            {
                int startIndex = brokerUri.IndexOf("failover:") + "failover:".Length;
                int endIndex = brokerUri.Contains("?") ? brokerUri.IndexOf("?") - "failover:".Length : brokerUri.Length;
                this.BrokerUriIpAddress = brokerUri.Substring(startIndex, endIndex).Replace("(", string.Empty).Replace(")", string.Empty);
            }
            else
            {
                int endIndex = brokerUri.Contains("?") ? brokerUri.IndexOf("?") : brokerUri.Length;
                this.BrokerUriIpAddress = brokerUri.Substring(0, endIndex);
            }
        }
        #endregion


        #region 委托方法
        /// <summary>
        /// 当连接恢复时发生。
        /// </summary>
        private void Connection_ConnectionResumedListener()
        {
            this.IsConnected = true;
            this.ConnectionStatusChanged?.Invoke(this, new StatusEventArgs(true));
        }

        /// <summary>
        /// 当连接断开时发生。
        /// </summary>
        private void Connection_ConnectionInterruptedListener()
        {
            this.IsConnected = false;
            this.ConnectionStatusChanged?.Invoke(this, new StatusEventArgs(false));
        }

        /// <summary>
        /// 当与ActiveMQ通讯过程中出现异常时发生。
        /// </summary>
        /// <param name="exception"></param>
        private void Connection_ExceptionListener(Exception exception)
        {
            if (exception.InnerException != null && exception.InnerException.InnerException != null)
            {
                if (exception.InnerException.InnerException is System.Net.Sockets.SocketException)
                {
                    System.Net.Sockets.SocketException ex = exception.InnerException.InnerException as System.Net.Sockets.SocketException;
                    if (ex.SocketErrorCode == System.Net.Sockets.SocketError.ConnectionReset)
                    {
                        this.Connection_ConnectionInterruptedListener();
                    }
                }
            }
            this.Error?.Invoke(this, new ExceptionEventArgs() { Exception = exception, Message = "ActiveMQ服务监听过程中发生错误。" });
        }
        #endregion


        #region 私有方法
        /// <summary>
        /// 创建生产者。
        /// </summary>
        /// <param name="destinationType">目标类型。</param>
        /// <param name="destinationName">目标名称。</param>
        /// <returns></returns>
        private IMessageProducer CreateProducer(DestType destinationType, string destinationName)
        {
            if (string.IsNullOrWhiteSpace(destinationName))
            {
                throw new ArgumentException("指定的参数 destinationName 为null或不合法。");
            }

            IMessageProducer producer = null;
            if (destinationType == DestType.Queue)//创建Queue的生产者
            {
                if (this.producerQueueDict.ContainsKey(destinationName) == false)
                {
                    IQueue queue = this.session.GetQueue(destinationName);
                    if (queue == null)
                    {
                        throw new NMSException("创建指定目标名称的 IQueue 生产者出错。");
                    }
                    producer = this.session.CreateProducer(queue);
                    this.producerQueueDict.TryAdd(destinationName, producer);
                }
                producer = this.producerQueueDict[destinationName];
            }
            else if (destinationType == DestType.Topic)//创建Topic的生产者
            {
                if (this.producerTopicDict.ContainsKey(destinationName) == false)
                {
                    ITopic topic = this.session.GetTopic(destinationName);
                    if (topic == null)
                    {
                        throw new NMSException("创建指定目标名称的 ITopic 生产者出错。");
                    }
                    producer = this.session.CreateProducer(topic);
                    this.producerTopicDict.TryAdd(destinationName, producer);
                }
                producer = this.producerTopicDict[destinationName];
            }
            return producer;
        }

        /// <summary>
        /// 创建消费者。
        /// </summary>
        /// <param name="destinationType">目标类型。</param>
        /// <param name="destinationName">目标名称。</param>
        /// <param name="ignoreExpiration">是否忽略过期。</param>
        /// <param name="isDurable">消息是否持久化。</param>
        /// <returns></returns>
        private IMessageConsumer CreateConsumer(DestType destinationType, string destinationName, bool ignoreExpiration = true, bool isDurable = false)
        {
            if (string.IsNullOrWhiteSpace(destinationName))
            {
                throw new ArgumentException("指定的参数 destinationName 为null或不合法。");
            }

            IMessageConsumer consumer = null;
            if (destinationType == DestType.Queue)//创建Queue类型的消费者
            {
                if (this.consumerQueueDict.ContainsKey(destinationName) == false)
                {
                    IQueue queue = this.session.GetQueue(destinationName);
                    if (queue == null)
                    {
                        throw new NMSException("创建指定目标名称的 IQueue 消费者出错。");
                    }
                    consumer = this.session.CreateConsumer(queue, null, isDurable);
                    MessageConsumer mc = (MessageConsumer)consumer;
                    mc.IgnoreExpiration = ignoreExpiration;
                    this.consumerQueueDict.TryAdd(destinationName, mc);
                }
                consumer = this.consumerQueueDict[destinationName];
            }
            else if (destinationType == DestType.Topic)//创建Topic类型的消费者
            {
                if (this.consumerTopicDict.ContainsKey(destinationName) == false)
                {
                    ITopic topic = this.session.GetTopic(destinationName);
                    if (topic == null)
                    {
                        throw new NMSException("创建指定目标名称的 ITopic 消费者出错。");
                    }
                    consumer = this.session.CreateConsumer(topic, null, isDurable);
                    MessageConsumer mc = (MessageConsumer)consumer;
                    mc.IgnoreExpiration = ignoreExpiration;
                    this.consumerTopicDict.TryAdd(destinationName, mc);
                }
                consumer = this.consumerTopicDict[destinationName];
            }
            return consumer;
        }

        /// <summary>
        /// 使用指定的消费者订阅消息。
        /// </summary>
        /// <param name="consumer">消费者对象。</param>
        /// <param name="handler">处理消息的委托。</param>
        private void SubscribeDestination(IMessageConsumer consumer, Action<IActivemqClient, DataEventArgs> handler)
        {
            if (consumer == null)
            {
                ExceptionEventArgs args = new ExceptionEventArgs()
                {
                    Exception = new ArgumentNullException("consumer"),
                    Message = "使用指定的消费者订阅消息时出错。无效的消费者"
                };
                this.Error?.Invoke(this, args);
                return;
            }


            Action<IActivemqClient, DataEventArgs> receiver = handler;
            consumer.Listener += (_msg) =>
            {
                if (receiver == null)
                {
                    return;
                }
                try
                {
                    ITextMessage textMessage = (ITextMessage)_msg;
                    DataEventArgs args = new DataEventArgs()
                    {
                        Text = textMessage.Text,
                        Message = _msg,
                        Bytes = this.Encoding.GetBytes(textMessage.Text)
                    };
                    receiver(this, args);
                }
                catch (Exception ex)
                {
                    this.Error?.Invoke(this, new ExceptionEventArgs()
                    {
                        Exception = ex,
                        Message = $"接收消息时发生错误。ActiveMQ客户端Id：{this.ClientId}。"
                    });
                }
            };
        }
        #endregion


        #region 公共方法，IActivemqClient成员
        /// <summary>
        /// 连接ActiveMQ服务。
        /// </summary>
        public bool Connect()
        {
            if (string.IsNullOrEmpty(this.BrokerUri))
            {
                throw new ArgumentNullException("属性 BrokerUri 不合法。必须指定一个合法的 BrokerUri 用于连接访问 ActiveMQ 服务。");
            }

            try
            {
                this.connectionFactory = new ConnectionFactory(this.BrokerUri);
                if (string.IsNullOrEmpty(this.UserName) || string.IsNullOrEmpty(this.Password))
                {
                    this.connection = this.connectionFactory.CreateConnection();
                }
                else
                {
                    this.connection = this.connectionFactory.CreateConnection(this.UserName, this.Password);
                }
                if (!string.IsNullOrEmpty(this.ClientId))
                {
                    this.connection.ClientId = this.ClientId;
                }
                this.connection.ConnectionInterruptedListener += this.Connection_ConnectionInterruptedListener;
                this.connection.ConnectionResumedListener += this.Connection_ConnectionResumedListener;
                this.connection.ExceptionListener += this.Connection_ExceptionListener;

                this.connection.Start();
                this.session = this.connection.CreateSession();
                this.IsConnected = this.connection.IsStarted;
                return true;
            }
            catch (Exception ex)
            {
                this.Error?.Invoke(this, new ExceptionEventArgs() { Exception = ex, Message = "连接ActiveMQ服务时发生错误。" });
                return false;
            }
        }

        /// <summary>
        /// 异步连接ActiveMQ服务。不可等待。
        /// </summary>
        public void AsyncConnect()
        {
            Task.Factory.StartNew(() =>
            {
                this.Connect();
            });
        }

        /// <summary>
        /// 异步连接ActiveMQ服务。不可等待。
        /// </summary>
        public void AsyncConnect(IEnumerable<string> dlls)
        {
            Task.Factory.StartNew(() =>
            {
                this.Connect();
                if (dlls != null)
                {
                    this.RegisterConsumer(dlls);
                }
                else
                {
                    this.RegisterConsumer();
                }
            });
        }

        /// <summary>
        /// 关闭连接，断开服务。
        /// </summary>
        public void Close()
        {
            foreach (IMessageProducer producer in this.producerQueueDict.Values)
            {
                if (producer != null)
                {
                    producer.Close();
                    producer.Dispose();
                }
            }
            foreach (IMessageProducer producer in this.producerTopicDict.Values)
            {
                if (producer != null)
                {
                    producer.Close();
                    producer.Dispose();
                }
            }
            foreach (IMessageConsumer consumer in this.consumerQueueDict.Values)
            {
                if (consumer != null)
                {
                    consumer.Close();
                    consumer.Dispose();
                }
            }
            foreach (IMessageConsumer consumer in this.consumerTopicDict.Values)
            {
                if (consumer != null)
                {
                    consumer.Close();
                    consumer.Dispose();
                }
            }

            this.producerQueueDict.Clear();
            this.producerTopicDict.Clear();
            this.consumerQueueDict.Clear();
            this.consumerTopicDict.Clear();

            //释放session,connection,connectionFactory
            this.session?.Close();
            this.session?.Dispose();
            if (this.connection != null)
            {
                if (this.connection.IsStarted)
                {
                    this.connection.Close();
                    this.connection.Dispose();
                }
            }
            this.connection = null;
            this.connectionFactory = null;

            this.ConnectionStatusChanged?.Invoke(this, new StatusEventArgs(false));
        }

        /// <summary>
        /// 注册消费者。
        /// </summary>
        public void RegisterConsumer()
        {
            Assembly[] assemblies = AppDomain.CurrentDomain.GetAssemblies();

            IEnumerable<string> files = Directory.GetFiles(AppDomain.CurrentDomain.BaseDirectory);
            files = files.Where(p => p.EndsWith(".dll") || p.EndsWith(".exe"));
            this.RegisterConsumer(files);
        }

        /// <summary>
        /// 注册消费者。
        /// </summary>
        /// <param name="dlls">要扫描的目标dll完整路径。</param>
        public void RegisterConsumer(IEnumerable<string> dlls)
        {
            if (!this.IsConnected)
            {
                return;
            }
            List<Assembly> assemblies = new List<Assembly>();
            foreach (string dll in dlls)
            {
                assemblies.Add(Assembly.LoadFile(dll));
            }

            foreach (Assembly assembly in assemblies)
            {
                Type[] typeConsumers = assembly.GetTypes()
                                               .Where(p => !p.IsAbstract && p.IsPublic && typeof(IConsumer).IsAssignableFrom(p))
                                               .ToArray();
                foreach (Type typeConsumer in typeConsumers)
                {
                    IConsumer consumer = (IConsumer)Activator.CreateInstance(typeConsumer);
                    this.RegisterConsumer(consumer);
                }
            }
        }

        /// <summary>
        /// 注册消费者。
        /// </summary>
        /// <param name="consumer"></param>
        /// <returns></returns>
        public bool RegisterConsumer(IConsumer consumer)
        {
            if (!consumer.IsSubscribed)
            {
                return false;
            }
            if (consumer.DestinationType == DestType.Queue && this.consumerQueueDict.ContainsKey(consumer.DestinationName))
            {
                throw new DuplicateInstanceException($"已经注册了目标类型为 {DestType.Queue} 目标名称为 {consumer.DestinationName} 的消费者。");
            }
            if (consumer.DestinationType == DestType.Topic && this.consumerTopicDict.ContainsKey(consumer.DestinationName))
            {
                throw new DuplicateInstanceException($"已经注册了目标类型为 {DestType.Topic} 目标名称为 {consumer.DestinationName} 的消费者。");
            }

            consumer.ActivemqClient = this;
            this.SubscribeDestination(consumer.DestinationType, consumer.DestinationName, consumer.Receive);
            return true;
        }

        /// <summary>
        /// 订阅消息。
        /// </summary>
        /// <param name="destinationType">目标类型。</param>
        /// <param name="destinationName">目标名称。</param>
        /// <param name="handler">回调的目标方法。</param>
        public void SubscribeDestination(DestType destinationType, string destinationName, Action<IActivemqClient, DataEventArgs> handler)
        {
            IMessageConsumer consumer = this.CreateConsumer(destinationType, destinationName);
            this.SubscribeDestination(consumer, handler);
        }

        /// <summary>
        /// 订阅消息。
        /// </summary>
        /// <param name="destinationType">目标类型。</param>
        /// <param name="destinationName">目标名称。</param>
        /// <param name="ignoreExpiration">是否忽视过期。</param>
        /// <param name="handler">回调的目标方法。</param>
        public void SubscribeDestination(DestType destinationType, string destinationName, bool ignoreExpiration, Action<IActivemqClient, DataEventArgs> handler)
        {
            IMessageConsumer consumer = this.CreateConsumer(destinationType, destinationName, ignoreExpiration, false);
            this.SubscribeDestination(consumer, handler);
        }

        /// <summary>
        /// 取消订阅。
        /// </summary>
        /// <param name="destinationType">目标类型。</param>
        /// <param name="destinationName">目标名称。</param>
        public void CancelSubscribeDestination(DestType destinationType, string destinationName)
        {
            IMessageConsumer consumer = null;
            if (destinationType == DestType.Queue)
            {
                if (this.consumerQueueDict.ContainsKey(destinationName))
                {
                    this.consumerQueueDict.TryRemove(destinationName, out consumer);
                }
            }
            else if (destinationType == DestType.Topic)
            {
                if (this.consumerTopicDict.ContainsKey(destinationName))
                {
                    this.consumerTopicDict.TryRemove(destinationName, out consumer);
                }
            }

            consumer?.Close();
            consumer?.Dispose();
        }

        /// <summary>
        /// 发送文本消息。
        /// </summary>
        /// <param name="content">消息内容。</param>
        /// <param name="destinationType">目标类型。</param>
        /// <param name="destinationName">目标名称。</param>
        public void Send(string content, DestType destinationType, string destinationName)
        {
            this.Send(content, destinationType, destinationName, false);
        }

        /// <summary>
        /// 发送文本消息。
        /// </summary>
        /// <param name="content">消息内容。</param>
        /// <param name="destinationType">目标类型。</param>
        /// <param name="destinationName">目标名称。</param>
        /// <param name="isPersistent">消息是否持久化。</param>
        public void Send(string content, DestType destinationType, string destinationName, bool isPersistent)
        {
            this.Send(content, destinationType, destinationName, isPersistent, null);
        }

        /// <summary>
        /// 发送文本消息。
        /// </summary>
        /// <param name="content">消息内容。</param>
        /// <param name="destinationType">目标类型。</param>
        /// <param name="destinationName">目标名称。</param>
        /// <param name="isPersistent">消息是否持久化。</param>
        /// <param name="msgProperties">消息附加属性。</param>
        public void Send(string content, DestType destinationType, string destinationName, bool isPersistent, IDictionary<string, string> msgProperties)
        {
            try
            {
                if (!this.IsConnected)
                {
                    this.Error?.Invoke(this, new ExceptionEventArgs()
                    {
                        Exception = new ConnectionFailedException("连接已断开"),
                        Message = "与ActiveMQ服务的连接已断开，无法发送消息。"
                    });
                    return;
                }
                IMessageProducer producer = this.CreateProducer(destinationType, destinationName);
                producer.RequestTimeout = new TimeSpan(0, 0, 2);
                if (producer != null)
                {
                    IMessage message = producer.CreateTextMessage(content);
                    if (msgProperties != null)
                    {
                        foreach (KeyValuePair<string, string> kv in msgProperties)
                        {
                            message.Properties.SetString(kv.Key, kv.Value);
                        }
                    }
                    MsgDeliveryMode mode = isPersistent == true ? MsgDeliveryMode.Persistent : MsgDeliveryMode.NonPersistent;
                    producer.Send(message, mode, MsgPriority.Normal, this.LiveTime);
                }
            }
            catch (Exception ex)
            {
                this.Error?.Invoke(this, new ExceptionEventArgs() { Exception = ex, Message = "发送文本消息失败。" });
            }
        }

        /// <summary>
        /// 发送字节数组消息。
        /// </summary>
        /// <param name="bytes">消息内容。</param>
        /// <param name="destinationType">目标类型。</param>
        /// <param name="destinationName">目标名称。</param>
        public void Send(byte[] bytes, DestType destinationType, string destinationName)
        {
            this.Send(bytes, destinationType, destinationName, false);
        }

        /// <summary>
        /// 发送字节数组消息。
        /// </summary>
        /// <param name="bytes">消息内容。</param>
        /// <param name="destinationType">目标类型。</param>
        /// <param name="destinationName">目标名称。</param>
        /// <param name="isPersistent">消息是否持久化。</param>
        public void Send(byte[] bytes, DestType destinationType, string destinationName, bool isPersistent)
        {
            this.Send(bytes, destinationType, destinationName, isPersistent, null);
        }

        /// <summary>
        /// 发送字节数组消息。
        /// </summary>
        /// <param name="bytes">消息内容。</param>
        /// <param name="destinationType">目标类型。</param>
        /// <param name="destinationName">目标名称。</param>
        /// <param name="isPersistent">消息是否持久化。</param>
        /// <param name="msgProperties">消息附加属性。</param>
        public void Send(byte[] bytes, DestType destinationType, string destinationName, bool isPersistent, IDictionary<string, string> msgProperties)
        {
            try
            {
                if (!this.IsConnected)
                {
                    this.Error?.Invoke(this, new ExceptionEventArgs()
                    {
                        Exception = new ConnectionFailedException("连接已断开"),
                        Message = "与ActiveMQ服务的连接已断开，无法发送消息。"
                    });
                    return;
                }
                IMessageProducer producer = this.CreateProducer(destinationType, destinationName);
                if (producer != null)
                {
                    IMessage message = producer.CreateBytesMessage(bytes);
                    if (msgProperties != null)
                    {
                        foreach (KeyValuePair<string, string> kv in msgProperties)
                        {
                            message.Properties.SetString(kv.Key, kv.Value);
                        }
                    }
                    MsgDeliveryMode mode = isPersistent == true ? MsgDeliveryMode.Persistent : MsgDeliveryMode.NonPersistent;
                    producer.Send(message, mode, MsgPriority.Normal, this.LiveTime);
                }
            }
            catch (Exception ex)
            {
                this.Error?.Invoke(this, new ExceptionEventArgs() { Exception = ex, Message = "发送字节数组消息失败。" });
            }
        }
        #endregion


        #region 析构函数
        /// <summary>
        /// 执行与释放或重置非托管资源相关的应用程序定义的任务。
        /// </summary>
        ~ActivemqClient()
        {
            this.Close();
        }
        #endregion


        #region 公共方法，IDisposable成员
        /// <summary>
        /// 执行与释放或重置非托管资源相关的应用程序定义的任务。
        /// </summary>
        public void Dispose()
        {
            this.Close();
        }
        #endregion
    }
}
